[fix][io] KCA sink: handle null values with KeyValue<Avro,Avro> schema #19861

nicoloboschi · 2023-03-20T10:26:35Z

Motivation

When using a KCA based sink connector with a topic with schema KeyValue<Avro, Avro>, if a message has a null value, the connector throws:

023-03-15T08:00:22,139+0000 [public/default/sink] ERROR org.apache.pulsar.io.kafka.connect.KafkaConnectSink - Error sending the record SinkRecord(sourceRecord=PulsarRecord(topicName=Optional[persistent://public/default/data-baselines.keyvalue_short_tests], partition=0, message=Optional[org.apache.pulsar.client.impl.MessageImpl@2ddf2db4], schema=KeyValueSchema(SEPARATED,org.apache.pulsar.client.impl.schema.generic.GenericAvroSchema@33df2fa0,org.apache.pulsar.client.impl.schema.generic.GenericAvroSchema@359ba6c9), failFunction=org.apache.pulsar.functions.source.PulsarSource$$Lambda$264/0x00000008404bb440@227fd2ce, ackFunction=org.apache.pulsar.functions.source.PulsarSource$$Lambda$263/0x00000008404bb040@7619e9b8), value=(key = "org.apache.pulsar.client.impl.schema.generic.GenericAvroRecord@3c717af4", value = "null"))
org.apache.kafka.connect.errors.DataException: Invalid null value for required STRUCT field
	at org.apache.pulsar.io.kafka.connect.schema.KafkaConnectData.defaultOrThrow(KafkaConnectData.java:385) ~[pulsar-io-kafka-connect-adaptor-2.10.3.5.jar:2.10.3.5]
	at org.apache.pulsar.io.kafka.connect.schema.KafkaConnectData.getKafkaConnectData(KafkaConnectData.java:66) ~[pulsar-io-kafka-connect-adaptor-2.10.3.5.jar:2.10.3.5]
	at org.apache.pulsar.io.kafka.connect.KafkaConnectSink.toSinkRecord(KafkaConnectSink.java:442) ~[pulsar-io-kafka-connect-adaptor-2.10.3.5.jar:2.10.3.5]
	at org.apache.pulsar.io.kafka.connect.KafkaConnectSink.write(KafkaConnectSink.java:125) ~[pulsar-io-kafka-connect-adaptor-2.10.3.5.jar:2.10.3.5]
	at org.apache.pulsar.functions.instance.JavaInstanceRunnable.sendOutputMessage(JavaInstanceRunnable.java:424) ~[?:?]
	at org.apache.pulsar.functions.instance.JavaInstanceRunnable.handleResult(JavaInstanceRunnable.java:394) ~[?:?]
	at org.apache.pulsar.functions.instance.JavaInstanceRunnable.run(JavaInstanceRunnable.java:335) ~[?:?]
	at java.lang.Thread.run(Thread.java:829) ~[?:?]

Modifications

Handle null value conversion to the kafka value

Verifying this change

Make sure that the change passes the CI checks.

Documentation

doc
doc-required
doc-not-needed
doc-complete

(cherry picked from commit f3747a1978c539c3f5a5f408988b9cc9125471a9)

eolivelli

Lgtm
Nice catch

dlg99 · 2023-03-20T16:26:14Z

Looking at this a bit deeper:
This should cover the specific issue we have right now.
However, I think the rootcause is that pulsar does not have "optional" marker in the schema.
In kafka's schema there is .isOptional(), in avro it is done via union with null type "type": [ "null", "string" ],
In pulsar it is lost.

So the fix will hack this for the top level key value schema, but the issue may reappear if the schema is generated for something like

Schema.KeyValue(Schema.INT32, Schema.KeyValue(Schema.INT32, Schema.KeyValue(Schema.INT32, schema)))

and the nested value is null.

We may need to remove "if (kafkaSchema.isOptional())" check in KafkaConnectData.defaultOrThrow()

eolivelli · 2023-03-20T16:32:08Z

actually in Pulsar a KeyValue schema allows null for both the key and the value.

@dlg99 in Pulsar it is not allowed to nest KeyValue schemas. KeyValue schema is a (bad) trick to provide a Schema to the key of the message.

any usage of nested KeyValue schemas is not supported (we may issue a warning or throw an error in Pulsar 3.0) and it cannot work properly

so this schema doesn't work
Schema.KeyValue(Schema.INT32, Schema.KeyValue(Schema.INT32, Schema.KeyValue(Schema.INT32, schema)))

dlg99 · 2023-03-20T16:35:47Z

actually in Pulsar a KeyValue schema allows null for both the key and the value.

@dlg99 in Pulsar it is not allowed to nest KeyValue schemas. KeyValue schema is a (bad) trick to provide a Schema to the key of the message.

any usage of nested KeyValue schemas is not supported (we may issue a warning or throw an error in Pulsar 3.0) and it cannot work properly

so this schema doesn't work Schema.KeyValue(Schema.INT32, Schema.KeyValue(Schema.INT32, Schema.KeyValue(Schema.INT32, schema)))

that's fine, nest any other pulsar schema.

The problem is in the loss of optional flag in nested schema in translation form kafka schema to pulsar (source -> topic -> transform -> sink chain)

...a-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSinkTest.java

dlg99 · 2023-03-20T21:48:57Z

...aptor/src/main/java/org/apache/pulsar/io/kafka/connect/schema/PulsarSchemaToKafkaSchema.java

@@ -80,6 +153,11 @@ private static org.apache.avro.Schema parseAvroSchema(String schemaJson) {
        return parser.parse(schemaJson);
    }

+    public static Schema getOptionalKafkaConnectSchema(org.apache.pulsar.client.api.Schema pulsarSchema) {
+        Schema s = getKafkaConnectSchema(pulsarSchema);
+        return new OptionalForcingSchema(s);


the only place it is used is within cache's get() call:

return schemaCache.get(pulsarSchema.getSchemaInfo().getSchema(), () -> { if (pulsarSchema.getSchemaInfo().getType() == SchemaType.KEY_VALUE) { KeyValueSchema kvSchema = (KeyValueSchema) pulsarSchema; return SchemaBuilder.map(getKafkaConnectSchema(kvSchema.getKeySchema()), getOptionalKafkaConnectSchema(kvSchema.getValueSchema())) .build(); }

eolivelli

LGTM

apache#19861) Co-authored-by: Andrey Yegorov <andrey.yegorov@datastax.com> (cherry picked from commit 03f8b80)

apache#19861) (#169) Co-authored-by: Andrey Yegorov <andrey.yegorov@datastax.com> (cherry picked from commit 03f8b80) Co-authored-by: Nicolò Boschi <boschi1997@gmail.com>

#19861) Co-authored-by: Andrey Yegorov <andrey.yegorov@datastax.com> (cherry picked from commit 03f8b80)

nicoloboschi added 2 commits March 20, 2023 11:21

[fix][io] KCA: handle null values with KeyValue<Avro,Avro> schema

5ebaf6c

(cherry picked from commit f3747a1978c539c3f5a5f408988b9cc9125471a9)

fix

f6faa66

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Mar 20, 2023

eolivelli approved these changes Mar 20, 2023

View reviewed changes

nicoloboschi added the ready-to-test label Mar 20, 2023

nicoloboschi self-assigned this Mar 20, 2023

nicoloboschi added this to the 3.0.0 milestone Mar 20, 2023

dlg99 reviewed Mar 20, 2023

View reviewed changes

...a-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSinkTest.java Show resolved Hide resolved

dlg99 added 2 commits March 20, 2023 12:02

handle null in kv's value with unwrap disabled

8cf537b

oops

c7de4f4

dlg99 reviewed Mar 20, 2023

View reviewed changes

eolivelli approved these changes Mar 20, 2023

View reviewed changes

dlg99 approved these changes Mar 20, 2023

View reviewed changes

dlg99 merged commit 03f8b80 into apache:master Mar 21, 2023

dlg99 pushed a commit to dlg99/pulsar that referenced this pull request Mar 21, 2023

[fix][io] KCA sink: handle null values with KeyValue<Avro,Avro> schema (

34c3ff0

apache#19861) Co-authored-by: Andrey Yegorov <andrey.yegorov@datastax.com> (cherry picked from commit 03f8b80)

dlg99 pushed a commit that referenced this pull request Mar 22, 2023

[fix][io] KCA sink: handle null values with KeyValue<Avro,Avro> schema (

71b4239

#19861) Co-authored-by: Andrey Yegorov <andrey.yegorov@datastax.com> (cherry picked from commit 03f8b80)

dlg99 added the cherry-picked/branch-2.11 label Mar 22, 2023

nicoloboschi deleted the kca-null-value branch March 23, 2023 11:13

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[fix][io] KCA sink: handle null values with KeyValue<Avro,Avro> schema #19861

[fix][io] KCA sink: handle null values with KeyValue<Avro,Avro> schema #19861

nicoloboschi commented Mar 20, 2023

eolivelli left a comment

dlg99 commented Mar 20, 2023

eolivelli commented Mar 20, 2023

dlg99 commented Mar 20, 2023 •

edited

Loading

dlg99 Mar 20, 2023 •

edited

Loading

eolivelli left a comment

[fix][io] KCA sink: handle null values with KeyValue<Avro,Avro> schema #19861

[fix][io] KCA sink: handle null values with KeyValue<Avro,Avro> schema #19861

Conversation

nicoloboschi commented Mar 20, 2023

Motivation

Modifications

Verifying this change

Documentation

eolivelli left a comment

Choose a reason for hiding this comment

dlg99 commented Mar 20, 2023

eolivelli commented Mar 20, 2023

dlg99 commented Mar 20, 2023 • edited Loading

dlg99 Mar 20, 2023 • edited Loading

Choose a reason for hiding this comment

eolivelli left a comment

Choose a reason for hiding this comment

dlg99 commented Mar 20, 2023 •

edited

Loading

dlg99 Mar 20, 2023 •

edited

Loading